package org.blame.confirm;


import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.blame.constants.Constants;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.sql.SQLOutput;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

public class PublishConfirm {

    private static final Integer MESSAGE_COUNT=1000;

    static Connection createConnection() throws IOException, TimeoutException {
        ConnectionFactory connectionFactory=new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT);
        connectionFactory.setUsername(Constants.USER_NAME);
        connectionFactory.setPassword(Constants.PASSWORD);
        connectionFactory.setVirtualHost(Constants.VIRTUAL);
        return connectionFactory.newConnection();
    }




    public static void main(String[] args) throws Exception {
//        单独确认
        publishingMessagesIndividually();

//        批量确认
        publishingMessagesInBatches();

//        异步确认
        handlingPublisherConfirmsAsynchronously();

    }

//    异步确认
    private static void handlingPublisherConfirmsAsynchronously()throws Exception {
        try(Connection connection=createConnection()) {
            Channel channel=connection.createChannel();
//            创建信道模式为confirm
            channel.confirmSelect();
            channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE3,true,false,false,null);
            long start=System.currentTimeMillis();
            SortedSet<Object> confirm = Collections.synchronizedSortedSet(new TreeSet<>());

            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliverTag, boolean multiple) throws IOException {
                    if(multiple){
                        confirm.headSet(deliverTag+1).clear();
                    }else{
                        confirm.remove(deliverTag);
                    }

                }

                @Override
                public void handleNack(long deliverTag, boolean multiple) throws IOException {
                    if(multiple){
                        confirm.headSet(deliverTag+1).clear();
                    }else{
                        confirm.remove(deliverTag);
                    }

                }
            });

//            发送消息
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String msg="confirm333"+i;
                long seqNo=channel.getNextPublishSeqNo();
                channel.basicPublish("",Constants.PUBLISHER_CONFIRMS_QUEUE3,null,msg.getBytes());
                confirm.add(seqNo);
            }
            while (!confirm.isEmpty()){
                Thread.sleep(10);
            }
            long end=System.currentTimeMillis();
            System.out.println("耗时"+(end-start));
        }
    }

//    批量确认
    private static void publishingMessagesInBatches()throws Exception {
        try(Connection connection=createConnection()) {
//            创建信道
            Channel channel=connection.createChannel();
//            创建信道为confirm模式
            channel.confirmSelect();
//            申明队列
            channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE2,true,false,false,null);

//            发送消息
            int batchSize=100;
            int outStandingMessageCount=0;
            long start=System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String msg="confirm222"+i;
                channel.basicPublish("",Constants.PUBLISHER_CONFIRMS_QUEUE2,null,msg.getBytes());
                if(outStandingMessageCount==batchSize){
                    channel.waitForConfirmsOrDie(5000);
                    outStandingMessageCount=0;
                }
                outStandingMessageCount++;
            }
            if(outStandingMessageCount>0){
                channel.waitForConfirmsOrDie(5000);
            }
            long end=System.currentTimeMillis();
            System.out.println("耗时"+(end-start));
        }
    }


    //   单独确认
    private static void publishingMessagesIndividually() throws Exception{
        try(Connection connection=createConnection()) {
//            创建信道
            Channel channel=connection.createChannel();

//            设置信道为confirm模式
            channel.confirmSelect();
//            声明信道
            channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE1,true,false,false,null);

//            发送消息
            String msg="单独确认进行消息的发送";
            long start=System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                System.out.println("confirms111"+i);

                channel.basicPublish("",Constants.PUBLISHER_CONFIRMS_QUEUE1,null,msg.getBytes(StandardCharsets.UTF_8));
                channel.waitForConfirmsOrDie(5000);
            }

            long end=System.currentTimeMillis();
            System.out.println("耗时"+(end-start));
        }
    }




}
